package org.fastsyncer.manager.framework;

import java.util.concurrent.ConcurrentLinkedQueue;

import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.common.exception.DataCenterException;
import org.fastsyncer.common.exception.TaskDistributorException;
import org.fastsyncer.manager.store.DataCenter;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 线程处理任务
 * 
 * @author YangPei
 * @date 2017年5月18日 下午10:41:03
 * @version 1.0.0
 */
public class TaskHandle extends Thread {

	private static Logger logger = LoggerFactory.getLogger(TaskHandle.class);

	private ManagerFactory managerFactory;

	private String name;

	public TaskHandle(ManagerFactory managerFactory, String name) {
		this.managerFactory = managerFactory;
		this.name = name;
	}

	private static DataCenter data = DataCenter.getInstance();

	public void run() {
		try {
			MappingTask mappingTask = (MappingTask) data.getMapping(name);
			if (mappingTask != null) {
				this.sync(mappingTask);
				logger.info("处理完成==============================>>" + name);
			}
		} catch (Exception e) {
			logger.error("执行全量同步时异常:" + e.getMessage());
		} finally {
			managerFactory.callback(name);
		}
	}

	private void sync(MappingTask mappingTask) throws Exception {
		// 总进度
		Integer process = mappingTask.getProcess();
		// 当前已完成进度
		Integer processCompleted = mappingTask.getProcessCompleted();
		long begin = System.currentTimeMillis();
		// 计算运行次数
		int runProcess = process - processCompleted;
		for (int k = 0; k < runProcess; k++) {
			// ########检测任务已被删除时，中止任务########
			String running = ManagerFactory.getRunning(mappingTask.getName());
			if (running == null) {
				throw new Exception("检测到任务" + mappingTask.getName() + "已被终止!");
			}

			// 1.计算拆分任务，返回待执行的任务范围[100000000,100000999],[100001000,100001999],...
			ConcurrentLinkedQueue<Integer[]> distributeTask = TaskDistributor.getInstance().distributeTask(mappingTask);
			if (distributeTask.isEmpty()) {
				throw new TaskDistributorException("拆分任务失败!");
			}

			// 2.数据分发开始
			JSONObject handle = RouterHandle.getInstance().router(mappingTask, distributeTask);
			// TODO 数据分发结束

			// 3.更新完成任务进度
			mappingTask.setProcessCompleted(processCompleted + 1);
			mappingTask.setSuccess(mappingTask.getSuccess() + handle.getInt("success"));
			mappingTask.setFail(mappingTask.getFail() + handle.getInt("fail"));
			mappingTask.setError(mappingTask.getError() + handle.getInt("error"));

			// 4.持久化任务信息
			boolean store = data.storeMapping(mappingTask.getName(), mappingTask);
			if (!store) {
				throw new DataCenterException("持久化任务存储失败!");
			}
			logger.info("刷新任务进度成功!\r");
		}

		// 更新最终成功失败数
		long runtime = System.currentTimeMillis() - begin;
		mappingTask.setRunTime(runtime);
		mappingTask.setFail(mappingTask.getSourceMapping().getRowCount() - mappingTask.getSuccess());
		data.storeMapping(mappingTask.getName(), mappingTask);
	}

}